package rabbitmq.test;

import com.rabbitmq.client.*;
import org.junit.Before;
import org.junit.Test;

import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.TimeoutException;

/**
 * 专门检测生产发送的逻辑
 */
public class MyProducer {
    //准备连接通道
    //每一个测试类都准备一个连接对象
    private Channel channel;
    @Before
    public void init(){
        //1.创建长连接 提供属性(ip,port,username,password,virtualHost)
        ConnectionFactory factory=new ConnectionFactory();
        factory.setHost("192.168.8.100");
        factory.setPort(5672);//TCP协议端口
        factory.setUsername("admin");
        factory.setPassword("123456");
        factory.setVirtualHost("/");
        try{
            Connection connection = factory.newConnection();
            //2.从长连接赋值短链接
            channel=connection.createChannel();
        }catch (Exception e){
            e.printStackTrace();
            System.out.println("创建连接失败");
        }
    }
    //发送简单的消息
    @Test
    public void sendSimpleMessage() throws Exception{
        //调用channel的api发送消息
        String message="你好rabbitmq,hello world";
        /**
         * String exchange: 生产者必须将消息发送到交换机,不能直接发到队列,指定交换机名称 使用的是默认交换机
         * String routingKey: 路由键,指定交换机将消息发送到哪个队列,使用默认交换机,路由键就是队列名称
         * BasicProperties props: 消息的属性,比如消息的优先级,是否持久化,消息的存活时间等
         * 优先级逻辑: 1000条消息 优先级都是1 有一个优先级是100 消费者接到1000个消息 先处理优先级高的
         * byte[] body: 消息体,就是发送的消息,如果发送内存数据 需要先进行序列化处理
         */
        channel.basicPublish(
                "",
                "queue01",
                null,
                message.getBytes(StandardCharsets.UTF_8));
        System.out.println("消息发送成功");
    }
    @Test
    public void sendTxMessage() throws IOException {
        String message="你好rabbitmq,hello world";
        try{
            //开启同步事务发送消息
            channel.txSelect();
            channel.basicPublish("","queue01",null,message.getBytes(StandardCharsets.UTF_8));
            //线程阻塞等待发送结果
            AMQP.Tx.CommitOk commitOk = channel.txCommit();
            System.out.println("本次发送事务结束,结果:{}"+commitOk);
        }catch (Exception e){
            //如果出现发送异常情况,比如连接超时,等待超时 关闭事务回滚
            channel.txRollback();
            System.out.println("消息发送失败,已回滚");
        }
    }
    //等待确认发布
    @Test
    public void sendConfirmMessage() throws IOException, InterruptedException, TimeoutException {
        //开启确认模式
        channel.confirmSelect();
        //正常发送消息
        String message="你好rabbitmq,hello world";
        channel.basicPublish("","queue01",null,message.getBytes(StandardCharsets.UTF_8));
        //设置超时等待时间(一旦超过时间上线 判断失败)
        boolean result = channel.waitForConfirms(500);//超过500毫秒就失败
        if (result){
            System.out.println("消息发送成功");
        }else{
            System.out.println("消息发送失败");
        }
    }
    //发送消息到路由交换机
    @Test
    public void sendDirectMessage() throws IOException {
        String message="消息";
        //携带 北京 路由key
        channel.basicPublish(
                "topic_ex",
                "中国.北京",null,message.getBytes());
    }

    @Test
    public void sendHeaderMessage() throws IOException {
        //在消息中携带头
        String message="消息";
        //携带消息的属性头
        Map<String, Object> headers = new HashMap<>();
        headers.put("user-id", "12345");
        headers.put("message-type", "order");
        headers.put("priority", 1);
        headers.put("timestamp", System.currentTimeMillis());

        // 5. 创建消息属性并设置头数据
        AMQP.BasicProperties properties = new AMQP.BasicProperties.Builder()
                .headers(headers)
                .contentType("text/plain")
                .deliveryMode(2) // 持久化消息
                .build();
        channel.basicPublish(
                "",
                "queue01",properties,message.getBytes());
    }
}
